package com.air.antispider.stream.dataprocess.businessprocess

import java.lang

import com.air.antispider.stream.commom.bean.ProcessedData
import com.air.antispider.stream.commom.util.jedis.{JedisConnectionUtil, PropertiesUtil}
import com.air.antispider.stream.commom.util.spark.SparkMetricsUtils
import com.alibaba.fastjson.JSONObject
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.json4s.DefaultFormats
import org.json4s.jackson.Json
import redis.clients.jedis.JedisCluster


object SparkStreamingMonitor {
  /**
    * spark任务监控
    * @param sc
    * @param processedDataRDD
    * @param serversCountMap
    */
  def streamMonitor(sc: SparkContext, processedDataRDD: RDD[ProcessedData], serversCountMap: collection.Map[String, Int]): Unit = {
    //应用ID
    val appId: String = sc.applicationId
    //应用名
    val appName: String = sc.appName

    //从spark监控网站中获取相应信息
    val url = "http://localhost:4040/metrics/json/"
    val jsonObj: JSONObject = SparkMetricsUtils.getMetricsJson(url)
    val gaugesObject: JSONObject = jsonObj.getJSONObject("gauges")
    val startKey = appId + ".driver." + appName + ".StreamingMetrics.streaming.lastCompletedBatch_processingStartTime"
    val endKey = appId + ".driver." + appName + ".StreamingMetrics.streaming.lastCompletedBatch_processingEndTime"

    //通过key查询出对应的开始结束时间
    val startTime: lang.Long = gaugesObject.getJSONObject(startKey).getLong("value")
    val endTime: lang.Long = gaugesObject.getJSONObject(endKey).getLong("value")

    //计算各项指标
    //时间差
    val costTime: Long = endTime - startTime
    //数量
    val sourceCount: Long = processedDataRDD.count()
    //速度
    val countPerMillis: Float = sourceCount.toFloat / costTime.toFloat

    //格式化结束时间的样式:yyyy-mm-dd hh:mm:ss
    val endTimeStr: String = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss").format(endTime)

    //将结果转为一个Map存入Redis中，便于前端识别
    val fieldMap = scala.collection.mutable.Map(
      "endTime" -> endTimeStr,
      "applicationUniqueName" -> appName,
      "applicationId" -> appId,
      "sourceCount" -> sourceCount.toString,
      "costTime" -> costTime.toString,
      "countPerMillis" -> countPerMillis.toString,
      "serversCountMap" -> serversCountMap)

    //将Map转为Json
    val value: String = Json(DefaultFormats).write(fieldMap)
    //存入Redis中
    val jedis: JedisCluster = JedisConnectionUtil.getJedisCluster
    val key: String = PropertiesUtil.getStringByKey("cluster.key.monitor.dataProcess", "jedisConfig.properties") + System.currentTimeMillis()
    //超时时间
    val ex:Int = PropertiesUtil.getStringByKey("cluster.exptime.monitor", "jedisConfig.properties").toInt
    //将数据存入Redis
    jedis.setex(key,ex,value)

    //将最后一批数据单独存入Redis
    val lastKey: String = PropertiesUtil.getStringByKey("cluster.key.monitor.dataProcess", "jedisConfig.properties") + "_LAST"
    jedis.set(lastKey,value)


  }

}
